#include "pch.h"
#include "FileReadTask.h"

#define CACHE_NUM	10

CFileReadTask::CFileReadTask(LPCWSTR lpFileName, IFileReaderPtr pCallback, INT64 nOffset, INT64 nLength)
	: m_bCanceled(false)
	, m_nCacheNum(0)
	, m_strFileName(lpFileName)
	, m_pCallback(std::move(pCallback))
	, m_nStartOffset(nOffset)
	, m_nRemainingSize(nLength)
	, m_nPerReadSize(0)
	, m_pTpIo(NULL)
	, m_overlapped{}
{
	AssertInThreadpool();

	LARGE_INTEGER liFileSize;
	FILETIME ftLastWrite;
	m_hFile = CreateFileW(lpFileName, GENERIC_READ, FILE_SHARE_READ | FILE_SHARE_DELETE, NULL, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OVERLAPPED, NULL);

	if (m_hFile == INVALID_HANDLE_VALUE)
		LogError(L"CreateFile error: %s, Error=%d", lpFileName, GetLastError());

	if (m_hFile == INVALID_HANDLE_VALUE ||
		m_pCallback->IsCanceled() ||
		!GetFileSizeEx(m_hFile, &liFileSize) ||
		!GetFileTime(m_hFile, NULL, NULL, &ftLastWrite))
		return;

	AT::OrderedTask(this, &IFileReader::OnOpen, m_pCallback, liFileSize.QuadPart, ftLastWrite);

	if (nLength == 0 || nOffset + nLength > liFileSize.QuadPart)
	{
		LogError(L"Read range error: %s, FileSize=%lld, Offset=%lld, Length=%lld", lpFileName, liFileSize.QuadPart, nOffset, nLength);
		return;
	}

	// 1/64 per read, min 64KB, max 1MB
	m_nPerReadSize = liFileSize.QuadPart >> 6;

	if (m_nPerReadSize < 64 * 1024)
		m_nPerReadSize = 64 * 1024;
	else if (m_nPerReadSize > 1024 * 1024)
		m_nPerReadSize = 1024 * 1024;
	else
		m_nPerReadSize = (m_nPerReadSize + 0xfff) & ~0xfff;

	m_pTpIo = CreateThreadpoolIo(m_hFile, [](PTP_CALLBACK_INSTANCE, PVOID pContext, PVOID, ULONG nIoResult, ULONG_PTR nBytesRead, PTP_IO)
	{
		((CFileReadTask *)pContext)->OnIoCompletion(nIoResult, nBytesRead);
	}, this, NULL);
}

CFileReadTask::~CFileReadTask()
{
	if (m_pTpIo)
		CloseThreadpoolIo(m_pTpIo);

	if (m_hFile != INVALID_HANDLE_VALUE)
		CloseHandle(m_hFile);

	if (m_nRemainingSize == 0)
		return;

	AT::OrderedTask(this, [](IFileReaderPtr pCallback, INT64 nOffset, INT64 nLength, LockExit lockExit)
	{
		lockExit = nullptr;	// Allow FileIOManager to exit
		pCallback->OnError(nOffset, nLength);
	}, std::move(m_pCallback), m_nStartOffset, m_nRemainingSize, m_lockExit);
}

void CFileReadTask::Start()
{
	AssertInThreadpool();

	if (m_pTpIo)
		DoRead();
}

void CFileReadTask::Cancel()
{
	m_bCanceled = true;

	if (m_pBuffer)
		CancelIoEx(m_hFile, NULL);
}

void CFileReadTask::DoRead()
{
	AssertInThreadpool();
	assert(!m_pBuffer);

	DWORD nBytesToRead = (DWORD)min(m_nRemainingSize, m_nPerReadSize);
	if (nBytesToRead)
		m_pBuffer.reset(malloc(nBytesToRead), free);

	if (!m_pBuffer)
		return;

	LARGE_INTEGER liOffset;
	liOffset.QuadPart = m_nStartOffset;
	m_overlapped.Offset = liOffset.LowPart;
	m_overlapped.OffsetHigh = liOffset.HighPart;

	StartThreadpoolIo(m_pTpIo);
	m_pLockThis = shared_from_this();	// Lock to prevent myself from being deleted during async IO

	DWORD dwError = 0;
	if (!ReadFile(m_hFile, m_pBuffer.get(), nBytesToRead, NULL, &m_overlapped) && (dwError = GetLastError()) != ERROR_IO_PENDING)
	{
		CancelThreadpoolIo(m_pTpIo);
		LogError(L"ReadFile error: %s, Error=%d", m_strFileName, dwError);
		m_pLockThis = nullptr;
	}
}

void CFileReadTask::OnIoCompletion(ULONG nIoResult, ULONG_PTR nBytesRead)
{
	AssertInThreadpool();

	if (nIoResult != NO_ERROR)
		LogError(L"ReadFile error: %s, IoResult=%d", m_strFileName, nIoResult);

	auto pThis = std::move(m_pLockThis);	// Unlock

	if (nIoResult == NO_ERROR && !m_bCanceled && !m_pCallback->IsCanceled())
	{
		AT::OrderedTask(this, &CFileReadTask::DoCallback, shared_from_this(), std::move(m_pBuffer), m_nStartOffset, nBytesRead);
		m_nStartOffset += nBytesRead;
		m_nRemainingSize -= nBytesRead;

		// Pause if there are too many caches
		if (++m_nCacheNum != CACHE_NUM)
			DoRead();
	}
}

void CFileReadTask::DoCallback(BufferPtr pBuffer, INT64 nOffset, size_t nLength)
{
	AssertInOrderedTask();

	if (!m_bCanceled && !m_pCallback->IsCanceled())
	{
		// Continue to read
		if (--m_nCacheNum == CACHE_NUM - 1)
			DoRead();

		m_pCallback->OnRead(std::move(pBuffer), nOffset, nLength);
	}
}
